package com.cskaoyan.mall.promo.mq;

import com.alibaba.fastjson.JSON;
import com.cskaoyan.mall.order.dto.CreateSeckillOrderRequest;
import com.cskaoyan.mall.promo.cache.CacheManager;
import com.cskaoyan.mall.promo.dal.persistence.PromoItemMapper;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @auther cskaoyan
 * @date 2022/6/27:16:00
 */
@Component
public class PromoOrderProducer {

    private TransactionMQProducer mqProducer;

    @Autowired
    PromoItemMapper promoItemMapper;

    @Autowired
    CacheManager cacheManager;
    @Autowired
    RedissonClient redissonClient;
    @PostConstruct
    public void init() {
        mqProducer = new TransactionMQProducer("promo_order_producer");

        // 设置nameserver地址
        mqProducer.setNamesrvAddr("127.0.0.1:9876");

        // 设置事物监听器
        mqProducer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                Map<String, Long> paramMap = (Map<String, Long>) arg;
                Long productId = paramMap.get("productId");
                Long psId = paramMap.get("psId");
                // 扣减秒杀商品库存
                Integer effectiveRow = 0;


                String lockKey = "promo_stock_lock " + productId + ":" + psId;
                // 从redis中获取一把锁
                RLock lockObj = redissonClient.getLock(lockKey);

                // 加锁
                //  对于我们从redis里获取到的一把普通分布式锁，如果加锁失败的线程，也会被阻塞
                lockObj.lock();
                try {
                    effectiveRow = promoItemMapper.decreaseStock(productId, psId);
                } finally {
                    // 释放锁
                    lockObj.unlock();
                }

                String localTransactionKey = "promo_local_transaction: " + msg.getTransactionId();
                String promoItemStockFlag = "promo_item_stock_not_enough:" + productId + ":" + psId;

                if (effectiveRow < 1) {
                    // 商品售罄，意味着 业务层面秒杀下单的上游服务执行失败(没有库存)
                    // 在redis中打上库存售罄的标志
                    cacheManager.setCache(promoItemStockFlag, "none", 1);
                    //将本地事物的执行结果存储到redis中
                    cacheManager.setCache(localTransactionKey, "fail", 1);

                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }

                cacheManager.setCache(localTransactionKey, "success", 1);
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                String localTransactionKey = "promo_local_transaction: " + msg.getTransactionId();

                String result = cacheManager.checkCache(localTransactionKey);
                if (result == null || result.trim().isEmpty()) {
                    // 没有获取到确切的事物的执行结果
                    return  LocalTransactionState.UNKNOW;
                }

                if ("success".equals(result)) {
                    // 本地事物执行成功
                    return LocalTransactionState.COMMIT_MESSAGE;
                }

                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        });

        try {
            // 启动producer
            mqProducer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }



    public boolean sendPromoOrderMessage(CreateSeckillOrderRequest request, Long productId, Long psId) {
        String requestJson = JSON.toJSONString(request);

        // 准备秒杀服务本地事务执行所需要的参数
        HashMap<String, Long> paramMap = new HashMap<>();
        paramMap.put("productId",productId);
        paramMap.put("psId", psId);

        // 1. 准备好待发送的事物消息
        Message message = new Message();
        message.setTopic("promo_order");
        message.setBody(requestJson.getBytes(Charset.forName("utf-8")));

        //2. 发送事物消息
        TransactionSendResult sendResult = null;
        try {
            // sendMessage的第二个参数，实际上是本地事物执行所需要的参数
           sendResult = mqProducer.sendMessageInTransaction(message, paramMap);
        } catch (MQClientException e) {
            e.printStackTrace();
        }


        if (sendResult != null && LocalTransactionState.COMMIT_MESSAGE.equals(sendResult.getLocalTransactionState())) {
            // 如果消息发送成功，且本地事务执行的状态也是成功的状态
            return true;
        }


        // 返回false  消息没发送成功，本地事务执行失败
        return false;

    }

}
